package com.dz.timer;

import com.dz.utils.ConcurrentDateUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName TimerEvent
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/14 18:16
 * @Version 1.0
 */
public class TimerEvent
{
    private static Logger logger = LoggerFactory.getLogger(TimerEvent.class);

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env =StreamExecutionEnvironment.createLocalEnvironment();

        env.addSource(new MySourceTuple3()).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple3<String, Long, Long>>() {
            private  long delay = 1000;

            private long currentMaxTimestamp;
            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentMaxTimestamp-delay);
            }

            @Override
            public long extractTimestamp(Tuple3<String, Long, Long> element, long previousElementTimestamp) {
                  long sourceTime = element.f2;
                currentMaxTimestamp = Math.max(System.currentTimeMillis(),currentMaxTimestamp);
               // logger.info("时间戳=={}=={}",sourceTime,ConcurrentDateUtil.dateFormat(sourceTime,null));

                return System.currentTimeMillis();
            }
        }).
                keyBy(new KeySelector<Tuple3<String, Long, Long>, String>() {
            @Override
            public String getKey(Tuple3<String, Long, Long> value) throws Exception {
                return value.f0;
            }
        }).process(new KeyedProcessFunction<String, Tuple3<String, Long, Long>, String>() {

            private boolean first =true;
            private Map<String,Long> cache = new ConcurrentHashMap<>();


            private void update(String key,Long value)
            {
                cache.put(key,value);
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                logger.info("定时器触发=={}=={}",timestamp, ConcurrentDateUtil.dateFormat(timestamp,null));
               synchronized (ctx)
               {   ctx.timerService().registerEventTimeTimer(timestamp);
                   String a = cache.toString();
                   out.collect(a);}

            }

            @Override
            public void processElement(Tuple3<String, Long, Long> value, Context ctx, Collector<String> out) throws Exception {

              synchronized (ctx)
              {
                  if(first)
                  {
                      first=false;

                      logger.info("第一次注册=={}",ConcurrentDateUtil.dateFormat(value.f2,null));

                      ctx.timerService().registerEventTimeTimer(value.f2+10000);
                      logger.info("第一次注册=={}",ConcurrentDateUtil.dateFormat(value.f2,null));
                  }
                  else
                  {
                      // logger.info("更新值");
                      //不是第一次
                      if(cache.containsKey(value.f0))
                      {
                          update(value.f0,value.f1);
                      }
                      else
                      {
                          cache.put(value.f0,value.f1);
                      }
                  }
              }

            }
        }).print("输出结果");


        env.execute("timer event test");

    }
}
